package com.zl.iot.docking.builder;

import com.zl.iot.docking.config.ServerInfo;
import com.zl.iot.docking.exception.ZLExcep;
import com.zl.iot.docking.model.AliConsume;
import com.zl.iot.docking.model.AliServerVO;
import com.zl.iot.docking.service.Consumer;
import org.apache.commons.codec.binary.Base64;
import org.apache.qpid.jms.JmsConnection;
import org.apache.qpid.jms.JmsConnectionListener;
import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
import org.slf4j.Logger;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.lang.reflect.InvocationTargetException;
import java.net.URI;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.List;
import java.util.stream.IntStream;


/**
 * @author chenzy
 * @since 2021-11-02
 **/
public class ConsumerBuilder<T extends Consumer> {
    private static ConsumerBuilder instance;
    private static final Logger logger = ServerInfo.getInfoLog();
    private AliServerVO aliServer;
    private List<T> consumers=new ArrayList<>();
    private final String clientId = "异步_" + System.currentTimeMillis();
    private final String signMethod = "hmacsha1";
    private String userName;
    private String password;
    private final Class<T> tClass;
    public  static synchronized <TT extends Consumer> ConsumerBuilder<TT> instance(Class<TT> tClass){
        if (instance==null){
            instance=new ConsumerBuilder<>(tClass);
        }
        return instance;
    }
    private ConsumerBuilder(Class<T> tClass) {
        this.tClass = tClass;
    }
    public void close() {
        consumers.forEach(consumer -> consumer.close());
    }
    public void buildList(AliConsume aliConsume,AliServerVO aliServer) {
//        logger.info("订阅:"+ JsonUtil.model2Str(aliConsume));
        logger.info("订阅:"+ aliConsume.getInstanceId()+"-"+aliConsume.getConsumeGroupId());
        var num=aliConsume.getConsumeNum();
        if (num<1){
            /*不需要创建订阅者*/
            return;
        }
        init(aliConsume,aliServer);
        IntStream.range(0,num).forEach(i->consumers.add(build()));
    }
    /*创建订阅者*/
    public T build() {
        var context=createContext(aliServer.getAmqHost());
        try {
            return tClass.getDeclaredConstructor(AliServerVO.class,Connection.class,Destination.class)
                    .newInstance(aliServer,getConnection(context,userName,password),getQueue(context));
        } catch (InstantiationException | IllegalAccessException | InvocationTargetException | NoSuchMethodException e) {
            throw new RuntimeException(e);
        }
//        return new Consumer(aliServer,getConnection(context,userName,password),getQueue(context));
    }


    public void init(AliConsume aliConsume,AliServerVO aliServer){
        var instanceId=aliConsume.getInstanceId();
        var consumeGroupId=aliConsume.getConsumeGroupId();
        this.aliServer= aliServer;

        var timeStamp = System.currentTimeMillis();
        var accessKey = aliServer.getAccessKeyId();
        var accessSecret = aliServer.getAccessSecret();
        userName = clientId + "|authMode=aksign"
                + ",signMethod=" + signMethod
                + ",timestamp=" + timeStamp
                + ",authId=" + accessKey
                + ",iotInstanceId=" + instanceId
                + ",consumerGroupId=" + consumeGroupId
                + "|";
        var signContent = "authId=" + accessKey + "&timestamp=" + timeStamp;
        password = doSign(signContent, accessSecret, signMethod);
    }
    private String doSign(String toSignString, String secret, String signMethod) {
        try {
            var mac = Mac.getInstance(signMethod);
            mac.init(new SecretKeySpec(secret.getBytes(), signMethod));
            return Base64.encodeBase64String(mac.doFinal(toSignString.getBytes()));
        } catch (NoSuchAlgorithmException | InvalidKeyException e) {
            throw ZLExcep.error( "创建阿里云订阅连接失败，生成用户密码错误",e);
        }
    }
    private Context createContext(String amqHost){
        String connectionUrl = "failover:(amqps://" + amqHost + ":5671?amqp.idleTimeout=80000)"
                + "?failover.reconnectDelay=30";
        var hashtable = new Hashtable<>();
        hashtable.put("connectionfactory.SBCF", connectionUrl);
        hashtable.put("queue.QUEUE", "default");
        hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
        try {
            return new InitialContext(hashtable);
        } catch (NamingException e) {
            throw ZLExcep.error("初始化连接工厂失败",e);
        }
    }
    private Connection getConnection(Context context, String userName, String password){
        try {
            ConnectionFactory cf = (ConnectionFactory) context.lookup("SBCF");
            var connection= cf.createConnection(userName, password);
            ((JmsConnection) connection).addConnectionListener(myJmsConnectionListener);
            return connection;
        } catch (NamingException | JMSException e) {
            throw ZLExcep.error("获取连接失败",e);
        }
    }
    private Destination getQueue(Context context){
        try {
            return (Destination) context.lookup("QUEUE");
        } catch (NamingException e) {
            throw ZLExcep.error("获取队列失败",e);
        }
    }
    private static JmsConnectionListener myJmsConnectionListener = new JmsConnectionListener() {
        /**
         * 连接成功建立。
         */
        @Override
        public void onConnectionEstablished(URI remoteURI) {
            logger.info("onConnectionEstablished, remoteUri:{}", remoteURI);
            logger.info("连接建立成功！");
        }

        /**
         * 尝试过最大重试次数之后，最终连接失败。
         */
        @Override
        public void onConnectionFailure(Throwable error) {
            logger.error("onConnectionFailure, {}", error.getMessage());
            logger.info("连接建立失败！");
        }

        /**
         * 连接中断。
         */
        @Override
        public void onConnectionInterrupted(URI remoteURI) {
            logger.info("onConnectionInterrupted, remoteUri:{}", remoteURI);
            logger.info("连接建立中断！");
        }

        /**
         * 连接中断后又自动重连上。
         */
        @Override
        public void onConnectionRestored(URI remoteURI) {
            logger.info("onConnectionRestored, remoteUri:{}", remoteURI);
            logger.info("连接重新建立！");
        }

        @Override
        public void onInboundMessage(JmsInboundMessageDispatch envelope) {
        }

        @Override
        public void onSessionClosed(Session session, Throwable cause) {
        }

        @Override
        public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {
        }

        @Override
        public void onProducerClosed(MessageProducer producer, Throwable cause) {
        }
    };

}
